package cluster

import (
	"context"
	"sync"
	"time"

	"gitee.com/doraemon1996/bunny/env"
	"gitee.com/doraemon1996/bunny/log"
	"gitee.com/doraemon1996/bunny/network"
	"gitee.com/doraemon1996/bunny/protocol"
	"gitee.com/doraemon1996/bunny/structure"
	"gitee.com/doraemon1996/bunny/utils"
)

// 节点消息接收器
type receiver struct {
	structure.Control

	// 所属节点
	node *cluster

	// 节点配置
	cfg *env.NodeConfig

	// 监听器
	ln network.Listener

	// 连接池
	mu      sync.Mutex
	connMap map[uint64]network.Conn

	pktMng protocol.PacketMng
}

func (r *receiver) Init(ctx context.Context, node *cluster, cfg *env.NodeConfig) (err error) {
	networkStr, addressStr, err := utils.ParseAddress(cfg.Listen)
	if err != nil {
		return
	}

	options := make([]network.Option, 2)
	options = append(options, network.WithNetwork(networkStr))
	options = append(options, network.WithNetwork(addressStr))

	r.Control.Init(ctx, func() {
		ln, lnErr := network.Listen(r.Context(), options...)
		if lnErr != nil {
			err = lnErr
			return
		}

		r.node = node
		r.cfg = cfg
		r.ln = ln
		r.pktMng = protocol.NewPacketMng()
		r.pktMng.Init(r.Context(), 1400, time.Second*30)
	})
	return
}

func (r *receiver) Start() {
	r.Control.Start(func() {
		r.pktMng.Start()
		r.ln.Start(r.OnConnect)
	})
}

func (r *receiver) Stop() {
	r.Control.Stop(func() {
		r.ln.Stop()

		r.mu.Lock()
		connMap := r.connMap
		r.connMap = nil
		r.mu.Unlock()

		for _, conn := range connMap {
			conn.Stop(nil)
		}

		r.pktMng.Stop()

		r.ln = nil
		r.cfg = nil
		r.node = nil
	})
}

// 新连接事件处理
func (r *receiver) OnConnect(conn network.Conn) {
	log.Infof("OnConnect from %v", conn)
	conn.Init(r.Context())

	if r.Closed() {
		log.Warnf("OnConnect from %v, but cluster closed and will stop it", conn)
		conn.Stop(nil)
		return
	}

	r.mu.Lock()
	defer r.mu.Unlock()

	if _, exist := r.connMap[conn.ID()]; exist {
		log.Errorf("OnConnect %v exist, will discard the new one", conn)
		conn.Stop(nil)
		return
	}

	r.connMap[conn.ID()] = conn

	conn.Start(nil)
}

// 消息事件处理
func (r *receiver) OnMessage(conn network.Conn, netMsg any) {

}
